package com.xiaoze.basistest;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *
 */
public class StreamWordCount {

    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 用parameter tool工具从程序启动参数中提取配置项
        // ParameterTool parameterTool = ParameterTool.fromArgs(args);

        // String host = parameterTool.get("host");
        // int port = parameterTool.getInt("port");

        // 从socket文本流读取数据
        // 使用nc -lk 7777命令，linux先输入这个命令，然后再启动程序
        DataStream<String> inputDataStream = env.socketTextStream("192.168.5.123",7777);

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = inputDataStream.flatMap(new MyFlatMapper())
                // 修改后，使用tuple.f0表示第0个位置的信息，将流按Tuple的第一个字段分区
                .keyBy(tuple -> tuple.f0)
                //累加流Tuple的第二个字段
                .sum(1)
                .setParallelism(3);

        sum.print();

        env.execute();

    }

}
